package org.myfram.flink.flinkonjar.domain.process.kafkaJoinHbase;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.myfram.flink.flinkonjar.domain.entity.KafkaJoinHBase.HBaseDimensionTable;
import org.myfram.flink.flinkonjar.domain.entity.KafkaJoinHBase.KafkaHBaseJoinedResult;
import org.myfram.flink.flinkonjar.domain.entity.KafkaJoinHBase.KafkaSourceUserEvent;

import java.util.HashMap;
import java.util.Map;

/**
     * 维表关联函数
     */
@Slf4j
public class DimensionJoinFunction extends CoProcessFunction<KafkaSourceUserEvent, HBaseDimensionTable, KafkaHBaseJoinedResult> {

    private Map<String, HBaseDimensionTable> dimensionCache = new HashMap<>();

    @Override
    public void processElement1(KafkaSourceUserEvent kafkaSourceUserEvent, CoProcessFunction<KafkaSourceUserEvent, HBaseDimensionTable, KafkaHBaseJoinedResult>.Context context, Collector<KafkaHBaseJoinedResult> collector) throws Exception {
        // 处理Kafka数据
        String id = kafkaSourceUserEvent.getId();
        HBaseDimensionTable dimensionData = dimensionCache.get(id);

        if (dimensionData != null) {
            // 找到匹配的维表数据，输出关联结果
            KafkaHBaseJoinedResult result = new KafkaHBaseJoinedResult();
            result.setId(id);
            result.setName(dimensionData.getName()); // 使用维表中的name
            collector.collect(result);
            log.info("维表关联成功 - Kafka ID: {}, 维表Name: {}", id, dimensionData.getName());
        } else {
            log.warn("未找到匹配的维表数据 - Kafka ID: {}", id);
        }
    }

    @Override
    public void processElement2(HBaseDimensionTable hBaseDimensionTable, CoProcessFunction<KafkaSourceUserEvent, HBaseDimensionTable, KafkaHBaseJoinedResult>.Context context, Collector<KafkaHBaseJoinedResult> collector) throws Exception {
        // 处理维表数据，更新缓存
        dimensionCache.put(hBaseDimensionTable.getId(), hBaseDimensionTable);
        log.info("更新维表缓存 - ID: {}, Name: {}", hBaseDimensionTable.getId(), hBaseDimensionTable.getName());
    }

}
